package com.lee.mq.mqserver.datacenter;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: lishuo
 * Date: 2023-10-24
 * Time: 17:49
 */

import com.lee.mq.common.MqException;
import com.lee.mq.mqserver.core.Binding;
import com.lee.mq.mqserver.core.Exchange;
import com.lee.mq.mqserver.core.MSGQueue;
import com.lee.mq.mqserver.core.Message;
import javafx.beans.binding.StringBinding;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个咧用来统一管理内存中的数据(交换机、队列、绑定、消息、队列和消息之间的绑定关系、未应答消息)
 */
public class MemoryDataCenter {
    // key 表示唯一的交换机名，value 表示交换机对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 表示唯一的队列名，value 表示队列对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 表示唯一交换机名，第二个 key 表示唯一的队列名
    private ConcurrentHashMap<String, ConcurrentHashMap<String,Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 表示唯一的消息 Id，value 表示消息对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 表示唯一队列名，value 表示队列中的消息链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个 key 表示唯一队列名，第二个 key 表示唯一消息 Id
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // 操作内存中的交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 交换机添加成功: exchangeName = "+exchange.getName());
    }
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功: exchangeName = "+exchangeName);
    }
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    // 操作内存中的队列
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 队列添加成功: queueName = "+queue.getName());
    }
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功: queueName = "+queueName);
    }
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    // 操作内存中的绑定
    public void insertBinding(Binding binding) throws MqException {
        // 1.根据交换机名查询是否存在对应的绑定，不存在则创建(本身是线程安全的)
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k->new ConcurrentHashMap<>());
        // 避免覆盖，这里需要加锁
        synchronized (bindingMap) {
            // 2.再根据队列名查询是否存在 binding 绑定
            if (bindingMap.get(binding.getQueueName()) != null) {
                // 该 binding 已存在，抛出异常
                throw new MqException("[MemoryDataCenter] 已存在绑定: queueName = "
                        +binding.getQueueName()+", exchangeName = "+binding.getExchangeName());
            }
            bindingMap.put(binding.getQueueName(),binding);
            System.out.println("[MemoryDataCenter] 绑定添加成功: queueName = "
                    +binding.getQueueName()+", exchangeName = "+binding.getExchangeName());
        }

    }
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            throw new MqException("[MemoryDataCenter] 该交换机没有任何绑定: "
                    +" exchangeName = "+binding.getExchangeName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功: queueName = "
                +binding.getQueueName()+", exchangeName = "+binding.getExchangeName());
    }
    // 获取绑定1：获取交换机的指定绑定
    public Binding getBinding(String exchangeName,String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }
    // 获取绑定2：获取交换机的所有绑定
    public ConcurrentHashMap<String, Binding> getBindings(String exchangName) {
        return bindingsMap.get(exchangName);
    }

    // 操作内存中的消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息添加成功: messageId = "+message.getMessageId());
    }
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功: messageId = "+messageId);
    }
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 操作内存中的队列和消息之间的绑定关系
    public void sendMessage(String queueName,Message message) {
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queueName, k -> new LinkedList<>());
        // LinkedList是线程不安全的，这里需要加锁
        synchronized (messages) {
            messages.add(message);
        }
        // 往 messageMap 中也插入一份
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息添加到队列成功: queueName = "+queueName
                +" messageId =" +message.getMessageId());
    }
    public Message pollMessage(String queueName) {
        // 头删，出队列
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            // 头删
            Message message = messages.remove(0);
            // messageMap 中也删除指定消息
            // messageMap.remove(message.getMessageId());
            System.out.println("[MemoryDataCenter] 消息重队列取出成功: queueName = "+queueName
                    +" messageId =" +message.getMessageId());
            return message;
        }
    }

    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    // 操作内存中的未应答消息

    // 添加未确定消息
    public void addMessageWaitAsk(String queueName,Message message) {
        ConcurrentHashMap<String, Message> messageWait = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageWait.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId=" + message.getMessageId());
    }
    // 删除未确认消息
    public void removeMessageWaitAsk(String queueName,String messageId) {
        ConcurrentHashMap<String, Message> messageWait = queueMessageWaitAckMap.get(queueName);
        if (messageWait == null) {
            return;
        }
        messageWait.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除待确认队列! messageId=" + messageId);
    }
    // 获取指定未确定消息
    public Message getMessageWaitAsk(String queueName,String messageId) {
        ConcurrentHashMap<String, Message> messageWait = queueMessageWaitAckMap.get(queueName);
        if (messageWait == null) {
            return null;
        }
        return messageWait.get(messageId);
    }

    // 服务器重启时，从硬盘上读取数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 0.清空所用数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();;
        messageMap.clear();
        queueMessageMap.clear();
        queueMessageWaitAckMap.clear();
        // 1.恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExcanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(),exchange);
        }
        // 2.恢复所有队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(),queue);
        }
        // 3.恢复所有绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
            bindingsMap.put(binding.getExchangeName(),bindingMap);
        }
        // 4.恢复所有消息数据
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromFile(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(),message);
            }
        }

        // 这里并不需要进行“未确认消息”的恢复，
        // 一旦在等待 ack 的过程中, 服务器重启了, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" .
    }

}
